package io.jagat.mapi;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Properties;

public class Main {
    
    public static void main(String[] args) throws Exception {
        Properties properties = new Properties();
        properties.put("runtime-config.timezone", "Asia/Shanghai");
        
        /*
         * # 为用户授权
         * grant all privileges on *.* to 'username'@'%' with grant option;
         * # 授权之后刷新权限
         * flush privileges;
         *
         */
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("172.16.1.11")
                .port(30006)
                .databaseList("test") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("test.tester") // 设置捕获的表
                .username("test")
                .password("85442791")
                .serverTimeZone("UTC") //必须与mysql服务端时区保持一致，
                //.jdbcProperties(properties)
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                .build();


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置 3s 的 checkpoint 间隔
        env.enableCheckpointing(3000);
        SingleOutputStreamOperator<String> outSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                // 设置 source 节点的并行度为 4
                .setParallelism(1);

        outSource.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
//                SinkFunction.super.invoke(value, context);
                System.out.println("sink："+value);
            }
        }).setParallelism(2);// 设置 sink 节点并行度为 1


        env.execute("Print MySQL Snapshot + Binlog");



        /*StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置 3s 的 checkpoint 间隔
        env.enableCheckpointing(3000);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env,EnvironmentSettings.newInstance().inStreamingMode().build());
        String mysqlSourceDDL = "create table mysql_source(\n" +
                "id String,\n" +
                "name String,\n" +
                "age Int,\n" +
                "PRIMARY KEY (id) NOT ENFORCED \n" +
                ") WITH (\n" +
                "'connector' = 'mysql-cdc', \n" +
                "'hostname' = 'localhost',\n" +
                "'port' = '3306',\n" +
                "'username' = 'root',\n" +
                "'password' = '', \n" +
                "'server-id' = '5401-6000', \n" +
                "'database-name' = 'cdc_test',\n" +
                "'table-name' = 'tableA',\n" +
                "'server-time-zone' = 'Asia/Shanghai',\n" +
                "'scan.startup.mode' = 'initial', \n" +
//                "'scan.startup.mode' = 'latest-offset', \n" +
                "'debezium.snapshot.locking.mode'='none'\n" +
                ")"
                ;
        System.out.println(mysqlSourceDDL);
        tEnv.executeSql(mysqlSourceDDL);
        tEnv.executeSql("select  * from mysql_source").print();*/
    }
}